package com.dominos.voldemortstress;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;

import voldemort.client.ClientConfig;
import voldemort.client.protocol.admin.AdminClient;
import voldemort.client.protocol.admin.AdminClientConfig;

public class VoldemortStress implements VoldemortStressMBean {

	ClientConfig clientConfig;
	StressParameters stressParameters;
	private VoldemortStoreClient voldemortStoreClient;
	List<WriteThread> writeThreadList = new ArrayList<WriteThread>();
	List<ReadThread> readThreadList = new ArrayList<ReadThread>();
	private VoldemortStoreClientMaker voldemortStoreClientMaker;
	long testStartTime;
	long testEndTime;
	Logger logger = Logger.getLogger(this.getClass());

	public VoldemortStress() {
		this(new VoldemortStoreClientFactory());
	}

	public VoldemortStress(VoldemortStoreClientMaker voldemortStoreClientMaker) {
		this.voldemortStoreClientMaker = voldemortStoreClientMaker;
	}

	public void setClientConfig(ClientConfig clientConfig) {
		this.clientConfig = clientConfig;

	}

	public ClientConfig getClientConfig() {
		return this.clientConfig;
	}

	public VoldemortStoreClient getStoreClient() {
		return voldemortStoreClient;
	}

	public void setStressParameters(StressParameters stressParameters) {
		this.stressParameters = stressParameters;

	}

	public StressParameters getStressParameters() {
		return this.stressParameters;
	}

	@Override
	public void connectToStore() {
		this.voldemortStoreClient = this.voldemortStoreClientMaker
				.createStoreClient(this.clientConfig, this.stressParameters
						.getStoreName());
	}

	@Override
	public void disconnectFromStore() {

		voldemortStoreClient.close();
		voldemortStoreClient = null;

	}

	@Override
	public void startStressTest() {

		testStartTime = new Date().getTime();
		if (this.voldemortStoreClient == null)
			this.connectToStore();

		if (stressParameters.getNumberofWriteThreads() > 0
				&& stressParameters.getSizeOfWriteData() > 0) {
			byte[] writeData = new byte[stressParameters.getSizeOfWriteData()];

			for (int writeByteIndex = 0; writeByteIndex < (stressParameters
					.getSizeOfWriteData()); writeByteIndex++) {
				writeData[writeByteIndex] = 3;
			}

			for (int writeThreadIndex = 0; writeThreadIndex < stressParameters
					.getNumberofWriteThreads(); writeThreadIndex++) {
				WriteThread writeThread = new WriteThread(voldemortStoreClient,
						writeThreadIndex, writeData, stressParameters
								.getFixedNumberOfWrites(), stressParameters
								.getNumberOfWarmupWrites());
				writeThreadList.add(writeThread);
				new Thread(writeThread).start();
			}

		}

		if (stressParameters.getNumberofReadThreads() > 0) {

			ExecutorService threadReadExecutor = Executors
					.newFixedThreadPool(stressParameters
							.getNumberofReadThreads());

			readThreadList.clear();

			for (int readThreadIndex = 0; readThreadIndex < stressParameters
					.getNumberofReadThreads(); readThreadIndex++) {
				ReadThread readThread = new ReadThread(voldemortStoreClient,
						readThreadIndex);
				readThreadList.add(readThread);
			}

			for (ReadThread readThread : readThreadList)
				threadReadExecutor.submit(readThread);
		}

	}

	@Override
	public void stopStressTest() throws InterruptedException, IOException {
		for (WriteThread writeThread : writeThreadList) {
			writeThread.quit();
			while (!writeThread.isQuit())
				Thread.sleep(10);
		}

		for (ReadThread readThread : readThreadList) {
			readThread.quit();
			while (!readThread.isQuit())
				Thread.sleep(10);
		}

		testEndTime = new Date().getTime();
		if (!this.stressParameters.getLogFilePath().isEmpty())
			writeTestToLogFile();
	}

	public void runStressTestForFixedNumberOfOperations()
			throws InterruptedException, IOException {
		this.startStressTest();
		for (WriteThread writeThread : writeThreadList) {
			while (!writeThread.isQuit())
				Thread.sleep(10);
		}

		testEndTime = new Date().getTime();
		if (!this.stressParameters.getLogFilePath().isEmpty())
			writeTestToLogFile();

	}

	void writeTestToLogFile() throws IOException {

		File file = new File(stressParameters.getLogFilePath());

		boolean fileIsNew = !file.exists();

		BufferedWriter logFileWriter = new BufferedWriter(
				new OutputStreamWriter(new FileOutputStream(file, true)));

		if (fileIsNew) {
			StringBuilder stringBuilder = new StringBuilder();
			stringBuilder.append("Comment");
			stringBuilder.append(",");
			stringBuilder.append("Max Connections Per Node: ");
			stringBuilder.append(",");
			stringBuilder.append("Max Threads: ");
			stringBuilder.append(",");
			stringBuilder.append("Max Queued Requests: ");
			stringBuilder.append(",");
			stringBuilder.append("Size Of Put Data");
			stringBuilder.append(",");
			stringBuilder.append("Number of Put Threads");
			stringBuilder.append(",");
			stringBuilder.append("Number of Get Threads");
			stringBuilder.append(",");
			stringBuilder.append("Total Test Duration (ms)");
			stringBuilder.append(",");
			stringBuilder.append("Number Of Puts");
			stringBuilder.append(",");
			stringBuilder.append("Number of Put Errors");
			stringBuilder.append(",");
			stringBuilder.append("Average Put Time (ms)");
			stringBuilder.append(",");
			stringBuilder.append("Put Throughput (ops/second)");
			stringBuilder.append(",");
			stringBuilder.append("Number of Gets");
			stringBuilder.append(",");
			stringBuilder.append("Number of Get Errors");
			stringBuilder.append(",");
			stringBuilder.append("Average Get Time (ms)");
			stringBuilder.append(",");
			stringBuilder.append("Get Throughput (ops/second)");
			logFileWriter.write(stringBuilder.toString());
			logFileWriter.newLine();
		}

		StringBuilder stringBuilder = new StringBuilder();
		stringBuilder.append(this.stressParameters.getLogComment());
		stringBuilder.append(",");
		stringBuilder.append(this.getClientConfigMaxConnectionsPerNode());
		stringBuilder.append(",");
		stringBuilder.append(this.getClientConfigMaxThreads());
		stringBuilder.append(",");
		stringBuilder.append(this.getClientConfigMaxQueuedRequests());
		stringBuilder.append(",");
		stringBuilder.append(this.stressParameters.getSizeOfWriteData());
		stringBuilder.append(",");
		stringBuilder.append(this.stressParameters.getNumberofWriteThreads());
		stringBuilder.append(",");
		stringBuilder.append(this.stressParameters.getNumberofReadThreads());
		stringBuilder.append(",");
		stringBuilder.append(testEndTime - testStartTime);
		stringBuilder.append(",");
		stringBuilder.append(this.getTotalNumberOfWrites());
		stringBuilder.append(",");
		stringBuilder.append(this.getTotalNumberOfWriteErrors());
		stringBuilder.append(",");
		stringBuilder.append(this.getAverageWriteTime());
		stringBuilder.append(",");
		stringBuilder.append(this.getWritesPerSecond());
		stringBuilder.append(",");
		stringBuilder.append(this.getTotalNumberOfReads());
		stringBuilder.append(",");
		stringBuilder.append(this.getTotalNumberOfReadErrors());
		stringBuilder.append(",");
		stringBuilder.append(this.getAverageReadTime());
		stringBuilder.append(",");
		stringBuilder.append(this.getReadsPerSecond());

		logFileWriter.write(stringBuilder.toString());
		logFileWriter.newLine();

		logFileWriter.flush();
		logFileWriter.close();
	}

	public int getNumberOfActiveWriteThreads() {
		int activeCount = 0;

		for (WriteThread writeThread : writeThreadList) {
			if (!writeThread.isQuit())
				activeCount++;
		}

		return activeCount;
	}

	public int getNumberOfActiveReadThreads() {
		int activeCount = 0;

		for (ReadThread readThread : readThreadList) {
			if (!readThread.isQuit())
				activeCount++;
		}

		return activeCount;
	}

	@Override
	public float getAverageWriteTime() {
		float totalDurationInMilliseconds = 0;
		float totalWriteCount = 0;
		float totalWriteErrorCount = 0;

		if (writeThreadList.size() == 0)
			return 0;
		for (WriteThread writeThread : writeThreadList) {
			totalWriteCount += writeThread.getWriteCount();
			totalWriteErrorCount += writeThread.getErrorCount();
			totalDurationInMilliseconds += writeThread.getDuration();
		}

		float averageWriteTimeMillisecondsPerWrite = totalDurationInMilliseconds
				/ (totalWriteCount + totalWriteErrorCount);

		return averageWriteTimeMillisecondsPerWrite;
	}

	public float getWritesPerSecond() {
		float totalDurationInMilliseconds = 0;
		float totalWriteCount = 0;
		float totalWriteErrorCount = 0;

		if (writeThreadList.size() == 0)
			return 0;
		for (WriteThread writeThread : writeThreadList) {
			totalWriteCount += writeThread.getWriteCount();
			totalWriteErrorCount += writeThread.getErrorCount();
			totalDurationInMilliseconds += writeThread.getDuration();
		}

		float writesPerSeond = ((totalWriteCount + totalWriteErrorCount)
				/ (totalDurationInMilliseconds) * 1000);

		return writesPerSeond;
	}

	@Override
	public int getTotalNumberOfWrites() {
		int totalNumberOfWrites = 0;

		for (WriteThread writeThread : writeThreadList) {
			totalNumberOfWrites += writeThread.getWriteCount();
		}
		return totalNumberOfWrites;
	}

	@Override
	public int getTotalNumberOfWriteErrors() {
		int totalNumberOfWriteErrors = 0;

		for (WriteThread writeThread : writeThreadList) {
			totalNumberOfWriteErrors += writeThread.getErrorCount();
		}
		return totalNumberOfWriteErrors;
	}

	@Override
	public void runStressTestForXSeconds(int seconds)
			throws InterruptedException, IOException {
		this.startStressTest();
		Thread.sleep(seconds * 1000);
		this.stopStressTest();
	}

	@Override
	public void runStressTestFor5Seconds() throws InterruptedException,
			IOException {
		this.runStressTestForXSeconds(5);
	}

	@Override
	public void runStressTestFor10Seconds() throws InterruptedException,
			IOException {
		this.runStressTestForXSeconds(10);
	}

	@Override
	public void clearStore() {
		AdminClientConfig adminClientConfig = new AdminClientConfig();
		AdminClient adminClient = new AdminClient(clientConfig
				.getBootstrapUrls()[0], adminClientConfig);

		int nodeID = 0;
		String storeName = this.stressParameters.getStoreName();
		adminClient.truncate(nodeID, storeName);

		adminClient.stop();
	}

	@Override
	public float getAverageReadTime() {
		float durationInMilliseconds = 0;
		float totalReadCount = 0;
		float totalReadErrorCount = 0;

		if (readThreadList.size() == 0)
			return 0;
		for (ReadThread readThread : readThreadList) {
			totalReadCount += readThread.getReadCount();
			totalReadErrorCount += readThread.getErrorCount();
			durationInMilliseconds += readThread.getDuration();
		}

		float averageReadTimeInMillisecondsPerRead = durationInMilliseconds
				/ (totalReadCount + totalReadErrorCount);

		return averageReadTimeInMillisecondsPerRead;

	}

	public float getReadsPerSecond() {
		float durationInMilliseconds = 0;
		float totalReadCount = 0;
		float totalReadErrorCount = 0;

		if (readThreadList.size() == 0)
			return 0;
		for (ReadThread readThread : readThreadList) {
			totalReadCount += readThread.getReadCount();
			totalReadErrorCount += readThread.getErrorCount();
			durationInMilliseconds += readThread.getDuration();
		}

		float readsPerSecond = ((totalReadCount + totalReadErrorCount) / durationInMilliseconds) * 1000;

		return readsPerSecond;

	}

	@Override
	public int getTotalNumberOfReads() {
		int totalNumberOfReads = 0;

		for (ReadThread readThread : readThreadList) {
			totalNumberOfReads += readThread.getReadCount();
		}
		return totalNumberOfReads;
	}

	@Override
	public int getTotalNumberOfReadErrors() {
		int totalNumberOfReadErrors = 0;

		for (ReadThread readThread : readThreadList) {
			totalNumberOfReadErrors += readThread.getErrorCount();
		}
		return totalNumberOfReadErrors;
	}

	@Override
	public int getClientConfigMaxConnectionsPerNode() {
		return this.clientConfig.getMaxConnectionsPerNode();
	}

	@Override
	public void setClientConfigMaxConnectionsPerNode(int maxConnectionsPerNode) {
		this.clientConfig.setMaxConnectionsPerNode(maxConnectionsPerNode);
	}

	@Override
	public int getClientConfigMaxThreads() {
		return this.clientConfig.getMaxThreads();
	}

	@Override
	public void setClientConfigMaxThreads(int maxThreads) {
		this.clientConfig.setMaxThreads(maxThreads);
	}

	@Override
	public int getClientConfigMaxQueuedRequests() {
		return this.clientConfig.getMaxQueuedRequests();
	}

	@Override
	public void setClientConfigMaxQueuedRequests(int maxQueuedRequests) {
		this.clientConfig.setMaxQueuedRequests(maxQueuedRequests);
	}

	@Override
	public boolean getClientConfigSocketKeepAlive() {
		return this.clientConfig.getSocketKeepAlive();
	}

	@Override
	public void setClientConfigSocketKeepAlive(boolean socketKeepAlive) {
		this.clientConfig.setSocketKeepAlive(socketKeepAlive);
	}

	public int getWarmupWriteCount() {
		int warmupWriteCount = 0;

		for (WriteThread writeThread : this.writeThreadList)
			warmupWriteCount += writeThread.getWarmupWriteCount();

		return warmupWriteCount;
	}

}
